package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.TableProcess;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    public TableProcessFunction(MapStateDescriptor<String,TableProcess> mapStateDescriptor){
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        String tableName = value.getString("table");
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(tableName);
        if (tableProcess!=null){
            JSONObject dataJsonObj = value.getJSONObject("data");
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table", sinkTable);
            dataJsonObj.put("type", value.getString("type"));
            out.collect(dataJsonObj);
        }
    }
    private void filterColumn(JSONObject dataJsonObj,String sinkColumns){
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        JSONObject jsonObj = JSON.parseObject(value);
        String op = jsonObj.getString("op");
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if ("d".equals(op)){
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            broadcastState.remove(before.getSourceTable());
        }else {
            //如果对配置表做除了删除外的其它操作(读取、添加、修改)，将配置信息放到或者更新到广播状态中
            TableProcess after = jsonObj.getObject("after", TableProcess.class);
            //获取业务数据库维度表表名
            String sourceTable = after.getSourceTable();
            //获取输出到phoenix中的表表名
            String sinkTable = after.getSinkTable();
            //获取表中字段
            String sinkColumns = after.getSinkColumns();
            //获取建表主键
            String sinkPk = after.getSinkPk();
            //获取建表扩展
            String sinkExtend = after.getSinkExtend();

            checkTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
            broadcastState.put(sourceTable, after);
        }
    }
    //拼接phoenix中维度表 建表语句，并提前建表
    private void checkTable(String tableName, String sinkColumns, String pk, String ext) {
        //空值处理
        if (pk == null) {
            pk = "id";
        }
        if (ext == null) {
            ext = "";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + EduConfig.PHOENIX_SCHEMA + "." + tableName + "(");
        String[] columnArr = sinkColumns.split(",");
        for (int i = 0; i < columnArr.length; i++) {
            String column = columnArr[i];
            if(pk.equals(column)){
                createSql.append(column + " varchar primary key");
            }else {
                createSql.append(column + " varchar ");
            }
            //除了最后一个字段，后面都需要加，
            if(i < columnArr.length - 1){
                createSql.append(",");
            }
        }
        createSql.append(") " + ext);
        System.out.println("在phoenix中执行的建表语句:" +createSql);

        PhoenixUtil.executeSql(createSql.toString());
    }
}
